package com.ibm.cps.processors;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.log4j.Logger;
import org.bson.types.ObjectId;
import org.joda.time.DateTime;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Throwables;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.processors.vector.SchemaUtility;
import com.ibm.cps.sql.expression.ExpressionNode;
import com.ibm.cps.sql.expression.FieldEvaluator;
import com.ibm.cps.sql.expression.StringLiteralNode;
import com.ibm.cps.sql.expression.ValueExpression;
import com.ibm.cps.sql.expression.function.meta.IgnoredInFunctionCall;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

@Info(Name=MessageFields.SCALAR_CALCULATION,Type=Info.ProcessorType.Other)
public class ScalarCalculationProcessor extends SchemaInheritiveProcessor {
	public static final String OUTPUT_FIELDS = "OUTPUT_FIELDS";
	public static final String OUTPUT_FIELD_TYPES = "OUTPUT_FIELD_TYPES";
	
	@ValueParam @IgnoredInFunctionCall
	private String outputFieldsJson;
	@ValueParam @IgnoredInFunctionCall
	private String outputFieldsTypeJson;
	
	private Map<String, ValueExpression> outputFields;
	
	private static final Logger logger = Logger.getLogger(ScalarCalculationProcessor.class);

	public ScalarCalculationProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
	}
	
	private Map<String, ValueExpression> parseOutputFields() throws CPSException {
		ObjectNode outputFields = (ObjectNode) JSONUtility.fromString(outputFieldsJson);
		ObjectNode outputFieldTypes = (ObjectNode) JSONUtility.fromString(outputFieldsTypeJson);
		Map<String, ValueExpression> result = new HashMap<>();
		Iterator<String> fieldNames = outputFields.fieldNames();
		while (fieldNames.hasNext()) {
			String name = fieldNames.next();
			String json = outputFields.get(name).asText();
			String fieldType = outputFieldTypes.get(name).asText();
			ValueExpression expression = (ValueExpression)JSONUtility.fromJsonToObjectWithClassName(json, fieldType);
			result.put(name, expression);
		}
		return result;
	}

	@Override
	protected List<NewField> defineFieldsToBeAdded() {
		try {
			this.outputFields = parseOutputFields();
			logger.debug("==============ScalarCalculation ============");
			logger.debug(outputFields);
		} catch (CPSException e) {
			e.printStackTrace();
		}

		
		List<NewField> newFields = new ArrayList<>();
		for (String outputFieldName : outputFields.keySet()) {
			String dataType;
			ValueExpression expression = outputFields.get(outputFieldName);
			if(expression instanceof StringLiteralNode){
				dataType = "string";
			}else{
				dataType = "double";
			}
			newFields.add(new NewField(outputFieldName, dataType));
		}
		return newFields;
	}

	@Override
	public void dispose() {
	}

	@Override
	protected Collection<ParsedDataMessage> runProcessorLogic(ITimeSeriesSchema parentSchema,
			ITimeSeriesSchema outputSchema, ParsedDataMessage message) throws CPSException {
		Collection<ParsedDataMessage> messages = new ArrayList<>();
		ParsedDataMessageBuilder outputBuilder = ParsedDataMessageBuilder.createInhertiveBuilder(outputSchema, message, parentSchema);
		
		logger.debug("==============ScalarCalculation input============");
		logger.debug(parentSchema.getTagNames());
		Set<String> involvedFields = new HashSet<>();
		for (String outputFieldName : outputFields.keySet()) {
			ExpressionNode outputExpression = outputFields.get(outputFieldName);
			involvedFields.addAll(outputExpression.collectVariableNames());
		}
		FieldEvaluator evaluator = SchemaUtility.createEvaluator(parentSchema, message, involvedFields);
		logger.debug("==============ScalarCalculation output============");
		for (String outputFieldName : outputFields.keySet()) {
			ValueExpression expression = outputFields.get(outputFieldName);
			Object value = expression.evaluate(evaluator);
			logger.debug("Expression: " + expression + " output: " + value);
			outputBuilder.append(outputFieldName, value);
		}
		
		messages.add(outputBuilder.toParsedDataMessage());
		return messages;
	}

	@Override
	public void initialize() throws CPSException {

	}

}
